package cn.monkey.transport.netty.server.codec;

import cn.monkey.transport.core.ExchangeManager;
import cn.monkey.transport.netty.NettyExchange;
import cn.monkey.transport.netty.utils.ChannelUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.TypeParameterMatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

/**
 * @param <PKG> package can not be release
 */
public abstract class ExchangeChannelInboundHandler<EXCHANGE extends NettyExchange, PKG> extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ExchangeChannelInboundHandler.class);
    protected final ExchangeManager<Channel, EXCHANGE> exchangeManager;
    protected final boolean autoRelease;
    private final TypeParameterMatcher matcher;

    public ExchangeChannelInboundHandler(ExchangeManager<Channel, EXCHANGE> exchangeManager,
                                         boolean autoRelease) {
        this.exchangeManager = exchangeManager;
        this.autoRelease = autoRelease;
        this.matcher = TypeParameterMatcher.find(this, ExchangeChannelInboundHandler.class, "PKG");
    }

    public ExchangeChannelInboundHandler(ExchangeManager<Channel, EXCHANGE> nettyExchangeManager) {
        this(nettyExchangeManager, true);
    }

    protected String key(ChannelHandlerContext ctx) {
        return ChannelUtil.getId(ctx.channel());
    }

    public boolean acceptInboundMessage(Object msg) throws Exception {
        return matcher.match(msg);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        boolean release = true;
        try {
            if (acceptInboundMessage(msg)) {
                @SuppressWarnings("unchecked")
                PKG imsg = (PKG) msg;
                channelRead0(ctx, imsg);
            } else {
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally {
            if (autoRelease && release) {
                ReferenceCountUtil.release(msg);
            }
        }
    }

    protected void channelRead0(ChannelHandlerContext ctx, PKG msg) throws Exception {
        EXCHANGE exchange;
        String key = this.key(ctx);
        try {
            exchange = this.exchangeManager.findOrCreate(key, ctx.channel());
            if (log.isDebugEnabled()) {
                log.debug("find or create exchange: {}", exchange.id());
            }
        } catch (Throwable e) {
            log.error("find or create exchange error, key: {}, channel: {}: \n", key, ctx.channel(), e);
            return;
        }
        Scheduler scheduler = exchange.getAttribute(Scheduler.class.getName());
        if (scheduler != null) {
            this.onSchedulerChannelRead(ctx, msg, scheduler);
        } else {
            super.channelRead(ctx, msg);
        }
    }

    private void onSchedulerChannelRead(ChannelHandlerContext ctx, PKG msg, Scheduler scheduler) {
        Mono.just(msg)
                .subscribeOn(scheduler)
                .subscribe(ctx::fireChannelRead);
    }
}
